package drds.server;

import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.util.Iterator;
import java.util.Map;
import java.util.Properties;
import java.util.Set;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

import org.apache.commons.lang3.exception.ExceptionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.util.concurrent.ListeningExecutorService;
import com.google.common.util.concurrent.MoreExecutors;

import drds.server.buffer.ByteBufferPool;
import drds.server.buffer.DirectByteBufferPool;
import drds.server.cache.CacheService;
import drds.server.common.$;
import drds.server.config.Config;
import drds.server.config.classloader.DynamicLoad;
import drds.server.config.model.DRDS;
import drds.server.config.table.structure.TableStructureDetector;
import drds.server.datasource.GroupDataSource;
import drds.server.datasource.IConnectionHolder;
import drds.server.memory.Memory;
import drds.server.net.Acceptor;
import drds.server.net.Reactors;
import drds.server.net.SessionContextFactory;
import drds.server.net.SessionContextGroupShareResources;
import drds.server.route.RouteService;
import drds.server.route.sequence.SequnceProcessor;
import drds.server.util.CurrentTimeMillis;
import drds.server.util.Executor;
import drds.server.util.ExecutorUtil;
import drds.server.util.LogBackConfigLoader;

/**
 * mysql版本只是是5.7以上,现在开发使用的数据库是mysql-installer-community-5.7.16.0
 * 
 * @author 曹正辉<330937205@qq.com>
 */
public class Server {
	public static final String PRETTIFY_LINE = "===========*===========*===========*===========*===========*===========";
	public static final String NAME = "DRDS";
	private static final long TIME_UPDATE_PERIOD = 20L;
	private static final long DEFAULT_OLD_CONNECTION_CLEAR_PERIOD = 5 * 1000L;

	private static final Server INSTANCE = new Server();
	private static final Logger LOGGER = LoggerFactory.getLogger(Server.class);
	private final RouteService routerService;
	private final CacheService cacheService;
	private Properties dnIndexProperties;

	// 全局序列号
	private final SequnceProcessor sequnceProcessor = new SequnceProcessor();
	private final DynamicLoad dynamicLoad;
	private volatile int nextProcessor;

	// System Buffer Pool Instance
	private ByteBufferPool byteBufferPool;

	/**
	 * 内存管理类
	 */
	private Memory memory = null;

	public static final Server getInstance() {
		return INSTANCE;
	}

	private final Config config;
	private final ScheduledExecutorService scheduledExecutorService;
	private final AtomicBoolean isOnline;
	private final long startupTime;
	private SessionContextGroupShareResources[] processors;
	private Executor businessExecutor;
	private Executor timerExecutor;
	private ListeningExecutorService listeningExecutorService;

	private Server() {

		// 读取文件配置
		this.config = new Config();

		// 定时线程池，单线程线程池
		scheduledExecutorService = Executors.newSingleThreadScheduledExecutor();

		/**
		 * <pre>
		 * 是否在线， 
		 * manager中有命令控制 
		 * | offline 
		 * | Change status to OFF  默认启动的时候是OFF
		 * | online 
		 * | Change status to ON
		 * </pre>
		 */
		this.isOnline = new AtomicBoolean(true);

		// 缓存服务初始化
		cacheService = new CacheService();

		// 路由计算初始化
		routerService = new RouteService(cacheService);

		// load datanode active index from properties
		dnIndexProperties = loadDnIndexProperties();

		// catlet加载器
		dynamicLoad = new DynamicLoad(DRDS.getDRDSHomePath() + File.separator + "catlet", config.getDRDS().getCatletClassCheckSeconds());

		// 记录启动时间
		this.startupTime = CurrentTimeMillis.currentTimeMillis();
	}

	public ByteBufferPool getByteBufferPool() {
		return byteBufferPool;
	}

	public Executor getTimerExecutor() {
		return timerExecutor;
	}

	public DynamicLoad getDynamicLoad() {
		return dynamicLoad;
	}

	public SequnceProcessor getSequnceProcessor() {
		return sequnceProcessor;
	}

	public Memory getMemory() {
		return memory;
	}

	public Config getConfig() {
		return config;
	}

	/**
	 * 在界面设置DRDS HOME_PATH
	 */
	@SuppressWarnings("unused")
	public void beforeStart() throws IOException, Exception {
		String home = DRDS.getDRDSHomePath();
		String configFilePath = null;
		if (configFilePath != null) {
			LogBackConfigLoader.load(configFilePath);
		}
		// ZkConfig.instance().initZk();
	}

	public void startup() throws IOException {
		LOGGER.info(PRETTIFY_LINE);
		DRDS drds = config.getDRDS();
		int processorCount = drds.getProcessors();

		int threadPoolSize = drds.getProcessorExecutor();
		processors = new SessionContextGroupShareResources[processorCount];
		// a page size
		int pageSize = drds.getBufferPageSize();
		// total page number
		short pageNum = drds.getBufferPageNum();
		// minimum allocation unit
		short blockSize = drds.getBufferBlockSize();

		int bufferPoolType = drds.getProcessorBufferPoolType();
		long totalBufferSize = 0;
		switch (bufferPoolType) {
		case 0:
			/**
			 * blockSize pageSize 页的大小pageNum 页的个数readByteBufferBlock
			 */
			byteBufferPool = new DirectByteBufferPool(pageNum, pageSize, blockSize, drds.getFrontSocketSoRcvbuf());

			totalBufferSize = pageSize * pageNum;
			break;
		case 1:
			/**
			 * todo 对应权威指南修改：
			 * 
			 * bytebufferarena由6个bytebufferlist组成， 这六个list有减少内存碎片的机制
			 * 每个bytebufferlist由多个bytebufferchunk组成， 每个list也有减少内存碎片的机制
			 * 每个bytebufferchunk由多个page组成， 平衡二叉树管理内存使用状态，计算灵活
			 * 
			 * 设置的pagesize对应bytebufferarena里面的
			 * 每个bytebufferlist的每个bytebufferchunk的buffer长度
			 * bufferPoolChunkSize对应每个bytebufferchunk的 每个page的长度
			 * bufferPoolPageNumber对应 每个bytebufferlist有多少个bytebufferchunk
			 */

			totalBufferSize = 6 * pageSize * pageNum;
			break;
		default:
			byteBufferPool = new DirectByteBufferPool(pageNum, pageSize, blockSize, drds.getFrontSocketSoRcvbuf());
			totalBufferSize = pageSize * pageNum;
		}

		/**
		 * Off Heap For Merge/Order/Group/Limit 初始化
		 */
		if (drds.getUseOffHeapForMerge() == 1) {
			try {
				memory = new Memory(drds, totalBufferSize);
			} catch (NoSuchFieldException | IllegalAccessException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		}
		businessExecutor = ExecutorUtil.create("business_executor", threadPoolSize);
		timerExecutor = ExecutorUtil.create("timer_executor", drds.getTimerExecutor());
		listeningExecutorService = MoreExecutors.listeningDecorator(businessExecutor);

		for (int i = 0; i < processors.length; i++) {
			processors[i] = new SessionContextGroupShareResources("x" + i, byteBufferPool, businessExecutor);
		}

		LOGGER.info("using nio network handler ");

		Reactors reactors = new Reactors(DirectByteBufferPool.THREAD_LOCAL_BYTEBUFFER_PREX + "NIOREACTOR", processors.length);
		Acceptor acceptor = new Acceptor(DirectByteBufferPool.THREAD_LOCAL_BYTEBUFFER_PREX + NAME + "Server", drds.getBindIp(), DRDS.DEFAULT_PORT, new SessionContextFactory(), reactors);
		acceptor.start();
		LOGGER.info(PRETTIFY_LINE);

		LOGGER.info("init " + GroupDataSource.class.getSimpleName());
		Map<String, GroupDataSource> groupDataSourceMap = config.getGroupDataSourceMap();

		for (GroupDataSource groupDataSource : groupDataSourceMap.values()) {
			String index = dnIndexProperties.getProperty(groupDataSource.getHostName(), "0");
			if (!"0".equals(index)) {
				LOGGER.info("init datahost: " + groupDataSource.getHostName() + "  to use datasource index:" + index);
			}
			groupDataSource.init(Integer.parseInt(index));
		}

		scheduledExecutorService.scheduleAtFixedRate(updateTime(), 0L, TIME_UPDATE_PERIOD, TimeUnit.MILLISECONDS);
		scheduledExecutorService.scheduleAtFixedRate(processorCheck(), 0L, drds.getProcessorCheckPeriod(), TimeUnit.MILLISECONDS);
		scheduledExecutorService.scheduleAtFixedRate(dataSourceOldConnectionHolderQueueClear(), 0L, DEFAULT_OLD_CONNECTION_CLEAR_PERIOD, TimeUnit.MILLISECONDS);
		scheduledExecutorService.schedule(catletClassClear(), 30000, TimeUnit.MILLISECONDS);

		if (drds.getCheckTableConsistency() == 1) {
			scheduledExecutorService.scheduleAtFixedRate(tableStructureCheck(), 0L, drds.getCheckTableConsistencyPeriod(), TimeUnit.MILLISECONDS);
		}

		if (drds.getUseGlobleTableCheck() == 1) { // 全局表一致性检测是否开启
			scheduledExecutorService.scheduleAtFixedRate(glableTableConsistencyCheck(), 0L, drds.getGlableTableCheckPeriod(), TimeUnit.MILLISECONDS);
		}

		// 定期清理结果集排行榜，控制拒绝策略
		scheduledExecutorService.scheduleAtFixedRate(resultSetMapClear(), 0L, drds.getClearBigSqLResultSetMapMs(), TimeUnit.MILLISECONDS);

		// new Thread(tableStructureCheck()).start();
		LOGGER.info(PRETTIFY_LINE);
	}

	private Runnable catletClassClear() {
		return new Runnable() {
			@Override
			public void run() {
				try {
					dynamicLoad.clearUnUsedClass();
				} catch (Exception e) {
					LOGGER.error(ExceptionUtils.getStackTrace(e));
				}
			};
		};
	}

	/**
	 * 清理 reload @@config_all 后，老的 connection 连接
	 */
	private Runnable dataSourceOldConnectionHolderQueueClear() {
		return new Runnable() {
			@Override
			public void run() {
				timerExecutor.execute(new Runnable() {
					@Override
					public void run() {

						long sqlTimeout = Server.getInstance().getConfig().getDRDS().getSqlExecuteTimeout() * 1000L;

						// 根据 lastTime 确认事务的执行， 超过 sqlExecuteTimeout 阀值 close
						// connection
						long currentTime = CurrentTimeMillis.currentTimeMillis();
						Iterator<IConnectionHolder> iterator = GroupDataSource.oldConnectionHolderQueue.iterator();
						while (iterator.hasNext()) {
							IConnectionHolder connectionHolder = iterator.next();
							long lastTime = connectionHolder.getLastTime();
							if (currentTime - lastTime > sqlTimeout) {
								connectionHolder.close("clear old backend connection ...");
								iterator.remove();
							}
						}
					}
				});
			};
		};
	}

	/**
	 * 在bufferpool使用率大于使用率阈值时不清理 在bufferpool使用率小于使用率阈值时清理大结果集清单内容
	 * 
	 */
	private Runnable resultSetMapClear() {
		return new Runnable() {
			@Override
			public void run() {
				try {
					ByteBufferPool byteBufferPool = getByteBufferPool();
					long size = byteBufferPool.size();
					long capacity = byteBufferPool.capacity();
					long bufferUsagePercent = (capacity - size) * 100 / capacity;
					if (bufferUsagePercent < config.getDRDS().getBufferUsagePercent()) {

					}
				} catch (Exception e) {
					LOGGER.error(ExceptionUtils.getStackTrace(e));
				}
			};
		};
	}

	private Properties loadDnIndexProperties() {
		Properties properties = new Properties();
		File file = new File(DRDS.getDRDSHomePath(), "conf" + File.separator + "dnindex.properties");
		if (!file.exists()) {
			return properties;
		}
		InputStream inputStream = null;
		try {
			inputStream = new FileInputStream(file);
			properties.load(inputStream);
		} catch (Exception e) {
			LOGGER.error(ExceptionUtils.getStackTrace(e));
			throw new RuntimeException(e);
		} finally {
			if (inputStream != null) {
				try {
					inputStream.close();
				} catch (IOException e) {
				}
			}
		}
		return properties;
	}

	/**
	 * save cur datanode index to properties file
	 * 
	 * @param
	 * @param currentActivedIndex
	 */
	public synchronized void saveDataHostIndex(String dataHost, int currentActivedIndex) {

		File file = new File(DRDS.getDRDSHomePath(), "conf" + File.separator + "dnindex.properties");
		FileOutputStream fileOutputStream = null;
		try {
			String oldIndex = dnIndexProperties.getProperty(dataHost);
			String newIndex = String.valueOf(currentActivedIndex);
			if (newIndex.equals(oldIndex)) {
				return;
			}

			dnIndexProperties.setProperty(dataHost, newIndex);
			LOGGER.info("save DataHost index  " + dataHost + " current index " + currentActivedIndex);

			File parent = file.getParentFile();
			if (parent != null && !parent.exists()) {
				parent.mkdirs();
			}

			fileOutputStream = new FileOutputStream(file);
			dnIndexProperties.store(fileOutputStream, "update");
		} catch (Exception e) {
			LOGGER.error($.Exception.getStackTrace("更新失败", e));
		} finally {
			if (fileOutputStream != null) {
				try {
					fileOutputStream.close();
				} catch (IOException e) {
				}
			}
		}

	}

	public RouteService getRouterService() {
		return routerService;
	}

	public CacheService getCacheService() {
		return cacheService;
	}

	public Executor getBusinessExecutor() {
		return businessExecutor;
	}

	public RouteService getRouterservice() {
		return routerService;
	}

	public SessionContextGroupShareResources nextProcessor() {
		int i = ++nextProcessor;
		if (i >= processors.length) {
			i = nextProcessor = 0;
		}
		return processors[i];
	}

	public SessionContextGroupShareResources[] getSessionContextGroupShareResources() {
		return processors;
	}

	public long getStartupTime() {
		return startupTime;
	}

	public boolean isOnline() {
		return isOnline.get();
	}

	public void offline() {
		isOnline.set(false);
	}

	public void online() {
		isOnline.set(true);
	}

	// 系统时间定时更新任务
	private Runnable updateTime() {
		return new Runnable() {
			@Override
			public void run() {
				CurrentTimeMillis.update();
			}
		};
	}

	// 处理器定时检查任务
	private Runnable processorCheck() {
		return new Runnable() {
			@Override
			public void run() {
				timerExecutor.execute(new Runnable() {
					@Override
					public void run() {
						try {
							for (SessionContextGroupShareResources processor : processors) {
								processor.checkJdbcConnectionHolder();
							}
						} catch (Exception e) {
							LOGGER.error(ExceptionUtils.getStackTrace(e));
						}

					}
				});
				timerExecutor.execute(new Runnable() {
					@Override
					public void run() {
						try {
							for (SessionContextGroupShareResources processor : processors) {
								processor.checkSessionContext();
							}
						} catch (Exception e) {
							LOGGER.error("checkFrontCons caught err:" + e);
						}
					}
				});
			}
		};
	}

	// 定时检查不同分片表结构一致性
	private Runnable tableStructureCheck() {
		return new TableStructureDetector();
	}

	// 全局表一致性检查任务
	private Runnable glableTableConsistencyCheck() {
		return new Runnable() {
			@Override
			public void run() {
				timerExecutor.execute(new Runnable() {
					@Override
					public void run() {
					}
				});
			}
		};
	}

	public ListeningExecutorService getListeningExecutorService() {
		return listeningExecutorService;
	}
}
